package cn.com.cennavi.services;

import java.io.Serializable;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;

import scala.Tuple2;
import cn.com.cennavi.bean.MineLabODBean;
import cn.com.cennavi.bean.TrajectoryBean;
import cn.com.cennavi.config.MineLabODConfig;

/**
 * 
 * @ClassName: DataParseService
 * @Description: TODO(数据解析模块)
 * @author buf
 * @date 2017年2月8日 上午9:35:56
 *
 */
public class DataParseService implements Serializable {
	private String dataSourse;

	public DataParseService(String dataSourse1) {
		dataSourse = dataSourse1;
	}

	/**
	 * 读取文件
	 */
	public JavaRDD<MineLabODBean> doJob(JavaRDD<String> filesData) {
		switch (dataSourse) {
		case "trajectory":
			JavaRDD<MineLabODBean> rdd = filesData.map(new Function<String, MineLabODBean>() {

				@Override
				public MineLabODBean call(String arg0) throws Exception {
					// TODO Auto-generated method stub
					try {
						TrajectoryBean tb = (TrajectoryBean) TrajectoryBean.parseData(arg0);
						if (tb == null) {
							return null;
						} else {
							MineLabODBean mb = new MineLabODBean();
							mb.oTime = tb.getOTime();
							mb.dTime = tb.getDTime();
							mb.oLongitude = tb.getOLongitude();
							mb.oLatitude = tb.getOLatitude();
							mb.dLongitude = tb.getDLongitude();
							mb.dLatitude = tb.getDLatitude();
							mb.dataSourse = dataSourse;
							System.out.println(mb.dataSourse + "," + dataSourse);
							return mb;
						}

					} catch (Exception e) {
						// TODO: handle exception
						e.printStackTrace();
					}
					return null;
				}
			});
			return rdd;
		default:
			// 扩展
			return null;
		}

	}

	/**
	 * 读取lzo文件
	 */
	public static JavaRDD<TrajectoryBean> doJob(JavaPairRDD<LongWritable, Text> lzofilesData) {
		JavaRDD<TrajectoryBean> objectData = lzofilesData.map(new Function<Tuple2<LongWritable, Text>, TrajectoryBean>() {

			/**
			 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么)
			 */
			private static final long serialVersionUID = 1L;

			public TrajectoryBean call(Tuple2<LongWritable, Text> arg0) throws Exception {
				// TODO Auto-generated method stub
				return null;
			}
		});
		return objectData;
	}
}
